feat(databricks): data quality monitoring from a contract's freshness and completeness SLOs - #864

…866)

Found while documenting #864 for forge_docs.

- Alerts: provider 1.140.0 does have notification_settings, but only
  under data_profiling_config (on_failure); anomaly_detection_config has
  only excluded_table_full_names. HONESTLY_TESTED said the provider had
  none.
- A refused monitor's hints do not come first: CLIError's three generic
  suggestions come before them, and they go ahead of the engine's own
  tofu state list hint. CHANGELOG, iac/base.py and the _apply_failure
  docstring now say that.
- The Free Edition quota covering the scans was assumed, not measured:
  the CHANGELOG now says to expect them to count against it.

…onitoring

#864 turns Databricks data quality monitoring on for an expose with an
anomaly_detection rule, and the monitor learns the rule's bounds, which
a dbt test cannot. The job's tests therefore leave such a rule to it
(databricks_quality.monitoring_request), as agreed with that change.
An expose that turns monitoring off (contract.dq.monitoring.enabled:
false) still has its anomaly rule refused, and the fix names the
opt-out. drift_detection stays refused.

…#871)

* feat(dq): read a contract's DQ rules once, as checks or refusals

fluid_build/util/dq_checks.py turns exposes[].contract.dq.rules into
engine-neutral checks (not_null, unique, accepted_values, regex, bounds,
reference, freshness, schema) and a refusal, with its fix, for each rule
no check can run as written. The allowed values of a valid_values rule
come from the selected column: its validationRules (enum 'A | B' or
'A,B', regex, range) or labels.references <product>.<column> through
consumes[]. Thresholds below 1.0 are shares computed as fluid test
computes them; critical/error map to error, warn/info to warn.

Shared with the Databricks DQ sessions (monitoring, results table,
Lakeflow expectations) as the one reading of the rules.

* feat(databricks): a dbt job runs its contract's DQ rules as dbt tests

Each check of the rules on the exposes a Databricks dbt build writes is a
singular dbt test, <test path>/fluid_dq/<model>/<check>.sql, uploaded
beside the project; the user's files upload byte for byte (dbt describes
a model in one properties entry only). dbt build runs each test after
its model, so the run page lists every rule by id and an error-severity
failure fails the run.

not_null, unique, accepted_values and relationships run dbt's built-in
test SQL through adapter.dispatch; regex, bounds, freshness, schema and
shares are forge-cli's own SQL after dbt_utils (recency,
not_null_proportion) and dbt_expectations (match_regex,
table_columns_to_match_set), both Apache-2.0, with no package to install
on serverless. Every value reaches the file as an escaped Jinja string
and every SQL literal through databricks_access.string_literal.

A rule no test can run, a Git-sourced build with rules, a model or a
referenced source the project lacks are refused at validate and plan.
The plan lists each job's tests (dq_tests); report() reads dbt's result
lines from a run's output.

* feat(dq): a reference to a product not consumed keeps no expose

A source-aligned product consumes nothing, so a labels.references to
another product's column no longer refuses in the shared reader:
Reference.expose_id is None and the engine decides whether it can find
that table (the Databricks dbt job cannot, and refuses). Two exposes of
one consumed product stay refused as ambiguous. Each refusal carries
the rule's type, so an engine that runs a type another way (Databricks
data quality monitoring for anomaly_detection) can tell its own.

* feat(databricks): the dbt job refuses an unconsumed reference; a run's report covers rules dbt stopped before

The dbt test of a reference reads the other product from the dbt source
the project declares for the expose it consumes, so a reference to a
product consumes[] does not list is refused there. report() says once
when the task's output holds no result of any rule's test (dbt stopped
before them). The job's dbt task key is a constant, DBT_TASK_KEY, which
a run's report finds the dbt output under.

* fix(databricks): the schema test refs its model while dbt parses

dbt records a test's parents while it parses, with execute false, and
dbt build --select +model+ selects tests by them. The schema test called
ref() only inside its execute guard, so it had no parent and was never
selected: live on Databricks Free Edition dbt found 11 tests and ran 10.
The relation is now set before the guard. Every kind is pinned to ref
its model at parse time.

* docs(databricks): changelog entry and the live row for DQ rules as dbt tests

The Unreleased entry says what runs for each rule type, what is refused,
and what fluid apply prints. HONESTLY_TESTED records the unit tiers and
the manual live run on Databricks Free Edition (2026-11-12): a seeded
bad row failed the run on its two error rules and warned on two, a clean
run passed all 11 tests, the next plan was empty, and everything was
swept; it also lists what was not run live.

* refactor(dq): smaller units in the rule reader and the test compiler; two hardening fixes

From the branch's security review and ripwire --quality-delta:

- The schema test escapes the expected type as it escapes the type the
  warehouse reports. It is unreachable today (every Databricks column
  type passes _column_type first), but no longer depends on it (LOW).
- A reference refusal quotes the labels.references parts it names (LOW).
- compile_tests runs in stages (a _Project with its paths and lazily
  read sources, _test_folder, _add_test); _declared_sources reads
  through _properties_docs and _source_tables; _Rule._allowed reads each
  validationRule kind in its own method; read_rules and has_rules share
  _parts; report's wording is _outcome.
- The job's tests ride on _Source, so _emit_job keeps its signature and
  its task dict stays one line.

* feat(dq): each check keeps its rule's severity, threshold and operator as written

DqCheck.rule_severity, rule_threshold and rule_operator carry the rule
as the contract wrote it, beside the two-level severity a dbt test runs
at and the share a lower threshold becomes: Lakeflow expectations fail
an update on critical and drop the row on error, and a results table
reports one row per rule.

* fix(databricks): refuse a test name the project uses and a selection that skips the model

From the diff's delivery review:

- A rule's test is named after its id, and dbt names a singular test once
  per project: a project test of the same name made the whole job fail to
  parse ("dbt found two data_tests with the name"). It is refused at plan.
- A build whose dbt selection provably leaves the model out would never
  run its tests: refused, with the model's fqn and path evaluated by
  databricks_retention.selects_model. A tag, config or state selection
  cannot be told here; the run's report names any test that did not run.
- A real dbt (dbt-core with dbt-duckdb, as CI installs) parses the upload
  and dbt ls --select +orders+ lists every rule's test; with the schema
  test's ref put back behind its execute guard, it fails.
- Wording: a share is fluid test's own only for completeness and
  uniqueness; the anomaly and drift refusal says what fluid test does
  with them; dq_checks is the reading for the engines that run the rules,
  this job its first.

* feat(databricks): fluid apply reports what dbt found for each contract rule after a run

Under --mode amend-and-build, once the job's run ends (success or
failure), the run report reads the dbt task's output and prints one line
per contract rule's test: pass, warn or fail with its failing-row count,
error, skip, or not in the output (with the note when Databricks kept only
its end), then the summary. The rules come from the contract alone
(databricks_dbt_tests.expected_tests); the task is found by DBT_TASK_KEY,
so the fluid_keys task beside it is never read as dbt's.

runs/get-output is read once per task and feeds the dbt summary, the
failed task's tail and the rules alike (_task_output). A rule report is
printed for the dbt task when a later task failed too: its tests ran.
Never fails a build: an unreadable output or contract only drops the
report.

* fix(dq): refuse a threshold or bound no float holds, and a range no value meets

From the diff's defect scan: an integer threshold of hundreds of digits
raised OverflowError out of fluid validate and plan, and so did a window
too long to count; an infinite or NaN threshold, or a range bound past a
float's range, reached the SQL as a bare 'inf'. Each is now a refusal
(_finite, and parse_range drops a non-finite bound). A range no value
can meet (between 100 and 0, > 5 and < 1, > 3 and <= 3) is refused
instead of failing every row of its test.

* fix(databricks): defect-scan fixes to the DQ tests and the keys task

- Only the exposes with rules are tested, so only their models are looked
  for: a sibling output with no rules (no model of its name, or left out
  of the selection) was refused.
- properties.model names the model of the one table a build writes; a
  build that writes several (no outputs writes every expose) tests each
  through the model named after its table.
- fluid validate, which reads the contract before {{ env.* }} is
  substituted, no longer refuses a model named by a placeholder; the
  apply writes those tests.
- The schema test spells a decimal as the table emitter does
  (_canonical_type): decimal(12) is decimal(12,0), as Databricks reports.
- _jinja_str keeps characters outside the BMP whole: ensure_ascii wrote
  a surrogate pair, which dbt cannot encode (one emoji in a rule id
  failed the whole project's parse).
- The test-name clash check also covers the project's own files under
  fluid_dq/, and a project YAML nested past the recursion limit is passed
  over like any file that does not parse.
- The fluid_keys task (#860) runs with run_if ALL_DONE: a failing rule
  fails dbt after its model replaced the table, which dropped its keys.

* docs(databricks): AGENTS.md and the live row for the DQ tests and their report

AGENTS.md says the job runs the contract's DQ rules as dbt tests and
where the reader and the renderer live; the changelog says the keys task
now runs whatever dbt's outcome; HONESTLY_TESTED counts the tests again
(48 + 54 + 4) and lists what the review waves added.

* feat(databricks): an anomaly_detection rule is left to data quality monitoring

#864 turns Databricks data quality monitoring on for an expose with an
anomaly_detection rule, and the monitor learns the rule's bounds, which
a dbt test cannot. The job's tests therefore leave such a rule to it
(databricks_quality.monitoring_request), as agreed with that change.
An expose that turns monitoring off (contract.dq.monitoring.enabled:
false) still has its anomaly rule refused, and the fix names the
opt-out. drift_detection stays refused.

* fix(databricks): refuse an unpaired surrogate; a placeholder selection waits for the apply

From the fix wave's re-check:

- Since the Jinja strings keep raw Unicode, an unpaired surrogate (YAML's
  "\ud83d" escape loads one) in a rule id, an enum value or a regex
  raised UnicodeEncodeError out of fluid plan. The reader refuses one with
  the control characters, in ids too, and a value from the project's own
  YAML (a source name) that UTF-8 cannot hold is a refusal of its own.
- fluid validate, before {{ env.* }} is substituted, refused a build whose
  properties.select still held a placeholder as leaving the model out; the
  selection is judged only once it can be read, at the apply.

* docs(databricks): the anomaly hand-off to monitoring, and the counts, in the changelog and the live row

* fix(databricks): refuse a test path no file can hold; monitoring takes over only the history refusal

From the round-2 re-check:

- An unpaired surrogate in dbt_project.yml's first test path (YAML's
  "\ud83d" escape loads one) reached the generated tests' paths and raised
  UnicodeEncodeError out of fluid plan. It is a test-path refusal now, and
  the message names the path escaped.
- On an expose Databricks data quality monitoring watches, every refusal of
  an anomaly_detection rule was dropped, so a bad severity or an id holding
  a control character passed silently. DqRefusal.needs_history marks the
  one refusal monitoring takes over (the rule needs the table's history);
  the others are refused as before.

* fix(databricks): a DQ refusal names a lone surrogate from the project escaped

The round-3 re-check found model-paths reaching the model refusal's text
raw, and a lone surrogate there (a YAML escape loads one) made fluid plan
fail on printing the refusal. Every DQ refusal is built in _refusal, which
now spells such a character as its backslash escape; a source or table
name from the project's YAML in a reference refusal is covered the same way.

* docs(databricks): the live row counts the round-3 tests and names the escaped refusal text

* fix(dq): a valid_values rule allows what fluid test allows

#867 landed fluid test's reading of a valid_values rule,
providers/quality_engine.py::value_domain: the values the rule itself
lists, else every declaration its column makes (the first enum, regex and
range validationRule, and labels.references), all of which must hold.
dq_checks had its own, which let the column's declarations win over the
rule's list, split an enum on | only when it held one, and read every
validationRule of a kind; the dbt tests could then check other values than
fluid test does.

dq_checks now takes the values, pattern and bounds from value_domain and
keeps its refusals on top: a range fluid test cannot read (it passes one
over) or no value can meet, a value or pattern holding a control
character, and a labels.references that consumes[] cannot resolve. Its own
parse_enum and parse_range are gone; the range grammar is quality_engine's.
The Tower Management pack still reads as 330 checks with none refused,
check for check as before, and the demo's dbt products as 230 tests.

* docs(dq): valid_values reads as fluid test's value_domain, in AGENTS.md, the changelog and the live row

* chore(secrets): baseline line numbers after the rebase on #865 and #869

* test(databricks): the schema test's expected rows join their parts with an explicit +